package com.alibaba.nacos.naming.core.v2.service.impl;

import com.alibaba.nacos.api.exception.NacosException;
import com.alibaba.nacos.api.exception.runtime.NacosRuntimeException;
import com.alibaba.nacos.api.naming.pojo.Instance;
import com.alibaba.nacos.common.notify.NotifyCenter;
import com.alibaba.nacos.naming.core.v2.ServiceManager;
import com.alibaba.nacos.naming.core.v2.client.Client;
import com.alibaba.nacos.naming.core.v2.client.manager.ClientManager;
import com.alibaba.nacos.naming.core.v2.event.client.ClientOperationEvent;
import com.alibaba.nacos.naming.core.v2.event.client.ClientOperationEvent.ClientRegisterServiceEvent;
import com.alibaba.nacos.naming.core.v2.event.client.ClientOperationEvent.ClientSubscribeServiceEvent;
import com.alibaba.nacos.naming.core.v2.event.client.ClientOperationEvent.ClientUnsubscribeServiceEvent;
import com.alibaba.nacos.naming.core.v2.event.metadata.MetadataEvent.InstanceMetadataEvent;
import com.alibaba.nacos.naming.core.v2.pojo.InstancePublishInfo;
import com.alibaba.nacos.naming.core.v2.pojo.Service;
import com.alibaba.nacos.naming.core.v2.service.ClientOperationService;
import com.alibaba.nacos.naming.pojo.Subscriber;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import static com.alibaba.nacos.api.exception.NacosException.INVALID_PARAM;
import static com.alibaba.nacos.naming.core.v2.event.client.ClientOperationEvent.*;
import static java.lang.String.format;

@Component
@RequiredArgsConstructor
@Slf4j
public class EphemeralClientOperationServiceImpl
        implements ClientOperationService {

    private final ClientManager clientManager;

    private final ServiceManager serviceManager;


    @Override
    public void registerInstance(Service service,
                                 Instance instance,
                                 String clientId) throws NacosException {
        Service usingService = serviceManager.getOrCreate(service);
        log.info("registerInstance clientId:{} namespace:{} groupServiceName:{}", clientId, usingService.getNamespace(), usingService.getGroupedServiceName());
        if (!usingService.isEphemeral()) {
            throw new NacosRuntimeException(INVALID_PARAM, format("Current service %s is persistent service, can't register ephemeral instance", usingService.getGroupedServiceName()));
        }

        Client client = clientManager.getClient(clientId);
        checkClientIsLegal(clientId, client);

        log.info("registerInstance clientId:{}", clientId);
        InstancePublishInfo instancePublishInfo = buildPublishInfo(instance);
        client.addServiceInstance(usingService, instancePublishInfo);
        client.setLastUpdatedTime();
        client.recalculateRevision();

        NotifyCenter.publishEvent(new ClientRegisterServiceEvent(usingService, clientId));
        NotifyCenter.publishEvent(new InstanceMetadataEvent(usingService, instancePublishInfo.getMetadataId(), false));
    }

    @Override
    public void deregisterInstance(Service service,
                                   Instance instance,
                                   String connectId) {
        Service usingService = serviceManager.getOrCreate(service);
        Client client = clientManager.getClient(connectId);
        checkClientIsLegal(connectId, client);

        InstancePublishInfo removed = client.removeServiceInstance(usingService);
        client.setLastUpdatedTime();
        client.recalculateRevision();

        if (removed != null) {
            NotifyCenter.publishEvent(new ClientDeregisterServiceEvent(usingService, connectId));
            NotifyCenter.publishEvent(new InstanceMetadataEvent(usingService, removed.getMetadataId(), true));
        }
    }

    private static void checkClientIsLegal(String clientId,
                                           Client client) {
        if (client == null) {
            throw new NacosRuntimeException(NacosException.CLIENT_DISCONNECT, String.format("Client [%s] connection already disconnect, can't register ephemeral instance.", clientId));
        }

        if (!client.isEphemeral()) {
            throw new NacosRuntimeException(NacosException.INVALID_PARAM, String.format("Current client [%s] is persistent client, can't register ephemeral instance.", clientId));
        }
    }

    @Override
    public void subscribeService(Service service,
                                 Subscriber subscriber,
                                 String connectId) {
        Service usingService = serviceManager.getOrCreate(service);
        Client client = clientManager.getClient(connectId);
        checkClientIsLegal(connectId, client);
        client.addServiceSubscriber(usingService, subscriber);
        client.setLastUpdatedTime();

        NotifyCenter.publishEvent(new ClientSubscribeServiceEvent(usingService, connectId));
    }

    @Override
    public void unsubscribeService(Service service,
                                   Subscriber subscriber,
                                   String connectId) {

        Service usingService = serviceManager.getOrCreate(service);
        Client client = clientManager.getClient(connectId);
        checkClientIsLegal(connectId, client);
        client.removeServiceSubscriber(usingService);
        client.setLastUpdatedTime();

        NotifyCenter.publishEvent(new ClientUnsubscribeServiceEvent(usingService, connectId));
    }
}
